package org.studiox.graph.janusgraph.v020;

import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang.StringUtils;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.janusgraph.core.*;
import org.janusgraph.core.schema.*;
import org.janusgraph.graphdb.database.management.ManagementSystem;
import org.slf4j.Logger;
import org.studiox.graph.common.GraphEdge;
import org.studiox.graph.common.GraphLineage;
import org.studiox.graph.common.GraphVertex;
import org.studiox.graph.common.exception.GraphException;
import org.studiox.graph.janusgraph.common.AbstractJanusGraph;
import org.studiox.graph.janusgraph.common.util.DateUtil;
import org.studiox.graph.janusgraph.common.util.StringUtil;

import java.util.*;
import java.util.concurrent.ExecutionException;

import static org.studiox.graph.common.GraphConstant.JanusGraphLabelPropertyKeyIndexConstant.*;

public class JanusGraphV020 extends AbstractJanusGraph {

  private Logger logger;

  private JanusGraphFactory.Builder builder;

  public JanusGraphV020(Logger logger) throws GraphException {
    this.logger = logger;
    builder = JanusGraphFactory.build();
    buildGraphConfig();
  }

  public JanusGraphV020(Configuration config, String hbaseTableName, Logger logger) {
    this.logger = logger;
    builder = JanusGraphFactory.build();
    buildGraphConfig(config, hbaseTableName);
  }

  @Override
  public void setBuilderConfig(String key, Object value) {
    builder.set(key, value);
  }

  private JanusGraph janusGraph;

  public JanusGraph getJanusGraph() {
    return janusGraph;
  }

  private JanusGraphManagement mgmt;

  @Override
  public boolean mgmtNotContainsVertexLabel(String vertexLabelName) {
    return !mgmt.containsVertexLabel(vertexLabelName);
  }

  @Override
  public void mgmtMakeVertexLabel(String vertexLabelName) {
    mgmt.makeVertexLabel(vertexLabelName).make();
  }

  @Override
  public boolean mgmtNotContainsEdgeLabel(String edgeLabelName) {
    return !mgmt.containsEdgeLabel(edgeLabelName);
  }

  @Override
  public void mgmtMakeEdgeLabel(String edgeLabelName) {
    mgmt.makeEdgeLabel(edgeLabelName).multiplicity(Multiplicity.MULTI).make();
  }

  @Override
  public boolean mgmtNotContainsPropertyKey(String propertyKeyName) {
    return !mgmt.containsPropertyKey(propertyKeyName);
  }

  @Override
  public void mgmtMakePropertyKey(String propertyKeyName, Class<?> zlass) {
    mgmt.makePropertyKey(propertyKeyName).dataType(zlass).cardinality(Cardinality.SINGLE).make();
  }

  @Override
  public boolean mgmtNotContainsGraphIndex(String indexName) {
    return !mgmt.containsGraphIndex(indexName);
  }

  @Override
  public void mgmtBuildVertexIndexOnlyWithOneKeyUnique(
      String indexName, String propertyKeyName, String labelName) {
    PropertyKey propertyKey = mgmt.getPropertyKey(propertyKeyName);
    VertexLabel vertexLabel = mgmt.getVertexLabel(labelName);
    JanusGraphIndex janusGraphIndex =
        mgmt.buildIndex(indexName, Vertex.class)
            .addKey(propertyKey)
            .indexOnly(vertexLabel)
            .unique()
            .buildCompositeIndex();
    mgmt.setConsistency(janusGraphIndex, ConsistencyModifier.LOCK);
  }

  @Override
  public void mgmtBuildVertexIndexWithOneKey(String indexName, String propertyKeyName) {
    PropertyKey propertyKey = mgmt.getPropertyKey(propertyKeyName);
    mgmt.buildIndex(indexName, Vertex.class).addKey(propertyKey).buildCompositeIndex();
  }

  @Override
  public void mgmtBuildEdgeIndexWithOneKey(String indexName, String propertyKeyName) {
    PropertyKey propertyKey = mgmt.getPropertyKey(propertyKeyName);
    mgmt.buildIndex(indexName, Edge.class).addKey(propertyKey).buildCompositeIndex();
  }

  @Override
  public void mgmtBuildEdgeIndexWithTwoKey(
      String indexName, String firstPropertyKeyName, String secondPropertyKeyName) {
    PropertyKey firstPropertyKey = mgmt.getPropertyKey(firstPropertyKeyName);
    PropertyKey secondPropertyKey = mgmt.getPropertyKey(secondPropertyKeyName);
    mgmt.buildIndex(indexName, Edge.class)
        .addKey(firstPropertyKey)
        .addKey(secondPropertyKey)
        .buildCompositeIndex();
  }

  @Override
  public void mgmtBuildEdgeIndexWithThreeKey(
      String indexName,
      String firstPropertyKeyName,
      String secondPropertyKeyName,
      String thirdPropertyKeyName) {
    PropertyKey firstPropertyKey = mgmt.getPropertyKey(firstPropertyKeyName);
    PropertyKey secondPropertyKey = mgmt.getPropertyKey(secondPropertyKeyName);
    PropertyKey thirdPropertyKey = mgmt.getPropertyKey(thirdPropertyKeyName);
    mgmt.buildIndex(indexName, Edge.class)
        .addKey(firstPropertyKey)
        .addKey(secondPropertyKey)
        .addKey(thirdPropertyKey)
        .buildCompositeIndex();
  }

  @Override
  public boolean mgmtNotContainsRelationIndex(String edgeLabelName, String relationIndexName) {
    EdgeLabel edgeLabel = mgmt.getEdgeLabel(edgeLabelName);
    return !mgmt.containsRelationIndex(edgeLabel, relationIndexName);
  }

  @Override
  public void mgmtBuildRelationIndexWithOneKey(
      String edgeLabelName, String relationIndexName, String propertyKeyName) {
    EdgeLabel edgeLabel = mgmt.getEdgeLabel(edgeLabelName);
    PropertyKey propertyKey = mgmt.getPropertyKey(propertyKeyName);
    mgmt.buildEdgeIndex(edgeLabel, relationIndexName, Direction.BOTH, propertyKey);
  }

  @Override
  public void mgmtBuildRelationIndexWithTwoKey(
      String edgeLabelName,
      String relationIndexName,
      String firstPropertyKeyName,
      String secondPropertyKeyName) {
    EdgeLabel edgeLabel = mgmt.getEdgeLabel(edgeLabelName);
    PropertyKey firstPropertyKey = mgmt.getPropertyKey(firstPropertyKeyName);
    PropertyKey secondPropertyKey = mgmt.getPropertyKey(secondPropertyKeyName);
    mgmt.buildEdgeIndex(
        edgeLabel, relationIndexName, Direction.BOTH, firstPropertyKey, secondPropertyKey);
  }

  @Override
  public void mgmtBuildRelationIndexWithThreeKey(
      String edgeLabelName,
      String relationIndexName,
      String firstPropertyKeyName,
      String secondPropertyKeyName,
      String thirdPropertyKeyName) {
    EdgeLabel edgeLabel = mgmt.getEdgeLabel(edgeLabelName);
    PropertyKey firstPropertyKey = mgmt.getPropertyKey(firstPropertyKeyName);
    PropertyKey secondPropertyKey = mgmt.getPropertyKey(secondPropertyKeyName);
    PropertyKey thirdPropertyKey = mgmt.getPropertyKey(thirdPropertyKeyName);
    mgmt.buildEdgeIndex(
        edgeLabel,
        relationIndexName,
        Direction.BOTH,
        firstPropertyKey,
        secondPropertyKey,
        thirdPropertyKey);
  }

  @Override
  public void updateGraphIndexWithReindex(String indexName)
      throws ExecutionException, InterruptedException {
    mgmt.updateIndex(mgmt.getGraphIndex(indexName), SchemaAction.REINDEX).get();
  }

  @Override
  public void updateRelationIndexWithReindex(String edgeLabelName, String relationIndexName)
      throws ExecutionException, InterruptedException {
    EdgeLabel edgeLabel = mgmt.getEdgeLabel(edgeLabelName);
    mgmt.updateIndex(mgmt.getRelationIndex(edgeLabel, relationIndexName), SchemaAction.REINDEX)
        .get();
  }

  @Override
  public void awaitGraphIndexStatusWithEnabled(String indexName) throws InterruptedException {
    ManagementSystem.awaitGraphIndexStatus(janusGraph, indexName)
        .status(SchemaStatus.ENABLED)
        .call();
  }

  @Override
  public void awaitRelationIndexStatusWithEnabled(String edgeLabelName, String relationIndexName)
      throws InterruptedException {
    ManagementSystem.awaitRelationIndexStatus(janusGraph, relationIndexName, edgeLabelName)
        .status(SchemaStatus.ENABLED)
        .call();
  }

  @Override
  public void mgmtCommit() {
    mgmt.commit();
  }

  @Override
  public void mgmtRollback() {
    mgmt.rollback();
  }

  @Override
  public void open() throws GraphException {
    logger.info("## JanusGraph start to create instance...");
    janusGraph = builder.open();
    mgmt = janusGraph.openManagement();
    super.open();
    logger.info("## JanusGraph create instance finish.");
  }

  @Override
  public void close() {
    if (null != janusGraph && janusGraph.isOpen()) {
      logger.info("## JanusGraph start to close...");
      try {
        janusGraph.close();
      } catch (Throwable t) {
        logger.info("## JanusGraph instance close failed, caused by {}", t.getMessage(), t);
        return;
      }
      logger.info("## JanusGraph instance closed.");
    }
  }

  private Long createJanusGraphVertex(String uniq, String type, Long createTime) {
    if (StringUtils.isBlank(uniq)) {
      return null;
    }
    try {
      JanusGraphVertex vertex = janusGraph.addVertex(LABEL_VERTEX);
      vertex.property(VERTEX_PROPERTY_UNIQ, uniq);
      if (StringUtils.isNotBlank(type)) {
        vertex.property(VERTEX_PROPERTY_TYPE, type);
      }
      vertex.property(VERTEX_PROPERTY_DELETED, false);
      if (null != createTime) {
        vertex.property(VERTEX_PROPERTY_CREATE_TIME, createTime);
      } else {
        vertex.property(VERTEX_PROPERTY_CREATE_TIME, DateUtil.getCurrentTimestamp());
      }
      janusGraph.tx().commit();
      return (Long) vertex.id();
    } catch (Throwable t) {
      logger.error(t.getMessage(), t);
      janusGraph.tx().rollback();
      return null;
    }
  }

  private boolean updateJanusGraphVertex(
      List<Vertex> vertices, String uniq, String type, Long createTime) {
    if (null == vertices || vertices.isEmpty()) {
      return false;
    }
    for (Vertex vertex : vertices) {
      try {
        boolean skip = true;
        if (StringUtils.isNotBlank(uniq)) {
          if (vertex.property(VERTEX_PROPERTY_UNIQ).isPresent()) {
            if (!vertex.value(VERTEX_PROPERTY_UNIQ).toString().equals(uniq)) {
              vertex.property(VERTEX_PROPERTY_UNIQ, uniq);
              skip = false;
            }
          } else {
            vertex.property(VERTEX_PROPERTY_UNIQ, uniq);
            skip = false;
          }
        }
        if (StringUtils.isNotBlank(type)) {
          if (vertex.property(VERTEX_PROPERTY_TYPE).isPresent()) {
            if (!vertex.value(VERTEX_PROPERTY_TYPE).toString().equals(type)) {
              vertex.property(VERTEX_PROPERTY_TYPE, type);
              skip = false;
            }
          } else {
            vertex.property(VERTEX_PROPERTY_TYPE, type);
            skip = false;
          }
        }
        if (null != createTime) {
          if (vertex.property(VERTEX_PROPERTY_CREATE_TIME).isPresent()) {
            if (Long.parseLong(vertex.value(VERTEX_PROPERTY_CREATE_TIME).toString())
                != createTime) {
              vertex.property(VERTEX_PROPERTY_CREATE_TIME, createTime);
              skip = false;
            }
          } else {
            vertex.property(VERTEX_PROPERTY_CREATE_TIME, createTime);
            skip = false;
          }
        }
        if (!skip) {
          janusGraph.tx().commit();
        }
      } catch (Throwable t) {
        logger.error(t.getMessage(), t);
        janusGraph.tx().rollback();
      }
    }
    return true;
  }

  private Vertex getJanusGraphVertexById(Long vertexId) {
    if (null == vertexId || vertexId <= 0) {
      return null;
    }
    return janusGraph.traversal().V(vertexId).next();
  }

  private List<Vertex> getJanusGraphVerticesByUniq(String uniq) {
    if (StringUtils.isBlank(uniq)) {
      return Collections.emptyList();
    }
    return janusGraph.traversal().V().has(VERTEX_PROPERTY_UNIQ, uniq).toList();
  }

  private List<Vertex> getJanusGraphVerticesByType(String type) {
    if (StringUtils.isBlank(type)) {
      return janusGraph.traversal().V().toList();
    }
    return janusGraph.traversal().V().has(VERTEX_PROPERTY_TYPE, type).toList();
  }

  @Override
  public Long createVertex(String uniq, String type, Long createTime) {
    List<Vertex> vertices = getJanusGraphVerticesByUniq(uniq);
    if (null != vertices && !vertices.isEmpty()) {
      return null;
    }
    return createJanusGraphVertex(uniq, type, createTime);
  }

  @Override
  public Long createVertexNotPreCheck(String uniq, String type, Long createTime) {
    return createJanusGraphVertex(uniq, type, createTime);
  }

  @Override
  public boolean updateVertex(String uniq, String type, Long createTime) {
    List<Vertex> vertices = getJanusGraphVerticesByUniq(uniq);
    if (null != vertices && !vertices.isEmpty()) {
      return updateJanusGraphVertex(vertices, null, type, createTime);
    }
    vertices = getJanusGraphVerticesByType(type);
    return updateJanusGraphVertex(vertices, null, type, createTime);
  }

  private boolean updateEdgeSourceTargetVertexUniq(String uniq) {
    if (StringUtils.isBlank(uniq)) {
      return false;
    }
    List<Edge> inEdges =
        janusGraph.traversal().V().has(VERTEX_PROPERTY_UNIQ, uniq).inE(LABEL_EDGE).toList();
    if (null != inEdges) {
      for (Edge edge : inEdges) {
        try {
          boolean skip = true;
          if (edge.property(EDGE_PROPERTY_TARGET_VERTEX_UNIQ).isPresent()) {
            if (!edge.value(EDGE_PROPERTY_TARGET_VERTEX_UNIQ).toString().equals(uniq)) {
              edge.property(EDGE_PROPERTY_TARGET_VERTEX_UNIQ, uniq);
              skip = false;
            }
          } else {
            edge.property(EDGE_PROPERTY_TARGET_VERTEX_UNIQ, uniq);
            skip = false;
          }
          if (!skip) {
            janusGraph.tx().commit();
          }
        } catch (Throwable t) {
          logger.error(t.getMessage(), t);
          janusGraph.tx().rollback();
        }
      }
    }
    List<Edge> outEdges =
        janusGraph.traversal().V().has(VERTEX_PROPERTY_UNIQ, uniq).outE(LABEL_EDGE).toList();
    if (null != outEdges) {
      for (Edge edge : outEdges) {
        try {
          boolean skip = true;
          if (edge.property(EDGE_PROPERTY_SOURCE_VERTEX_UNIQ).isPresent()) {
            if (!edge.value(EDGE_PROPERTY_SOURCE_VERTEX_UNIQ).toString().equals(uniq)) {
              edge.property(EDGE_PROPERTY_SOURCE_VERTEX_UNIQ, uniq);
              skip = false;
            }
          } else {
            edge.property(EDGE_PROPERTY_SOURCE_VERTEX_UNIQ, uniq);
            skip = false;
          }
          if (!skip) {
            janusGraph.tx().commit();
          }
        } catch (Throwable t) {
          logger.error(t.getMessage(), t);
          janusGraph.tx().rollback();
        }
      }
    }
    return true;
  }

  @Override
  public boolean updateVertex(String oldUniq, String newUniq, String type, Long createTime) {
    if (StringUtils.isBlank(oldUniq) || StringUtils.isBlank(newUniq)) {
      return false;
    }
    List<Vertex> vertices = getJanusGraphVerticesByUniq(newUniq);
    if (null != vertices && !vertices.isEmpty()) {
      return false;
    }
    vertices = getJanusGraphVerticesByUniq(oldUniq);
    if (null == vertices || vertices.isEmpty()) {
      return false;
    }
    boolean updateVertexFlag = updateJanusGraphVertex(vertices, newUniq, type, createTime);
    boolean updateEdgeFlag = false;
    if (updateVertexFlag) {
      updateEdgeFlag = updateEdgeSourceTargetVertexUniq(newUniq);
    }
    return updateVertexFlag & updateEdgeFlag;
  }

  private void updateJanusGraphVertexDeletedStatus(List<Vertex> vertices, boolean deleted) {
    if (null == vertices || vertices.isEmpty()) {
      return;
    }
    for (Vertex vertex : vertices) {
      try {
        boolean skip = true;
        if (vertex.property(VERTEX_PROPERTY_DELETED).isPresent()) {
          boolean current = Boolean.parseBoolean(vertex.value(VERTEX_PROPERTY_DELETED).toString());
          if (current != deleted) {
            vertex.property(VERTEX_PROPERTY_DELETED, deleted);
            skip = false;
          }
        } else {
          vertex.property(VERTEX_PROPERTY_DELETED, deleted);
          skip = false;
        }
        if (!skip) {
          janusGraph.tx().commit();
        }
      } catch (Throwable t) {
        logger.error(t.getMessage(), t);
        janusGraph.tx().rollback();
      }
    }
  }

  private boolean updateJanusGraphVertexDeletedStatus(String uniq, String type, boolean deleted) {
    if (StringUtils.isNotBlank(uniq)) {
      List<Vertex> vertices = getJanusGraphVerticesByUniq(uniq);
      if (null != vertices && !vertices.isEmpty()) {
        updateJanusGraphVertexDeletedStatus(vertices, deleted);
      }
    } else {
      List<Vertex> vertices = getJanusGraphVerticesByType(type);
      updateJanusGraphVertexDeletedStatus(vertices, deleted);
    }
    return true;
  }

  @Override
  public boolean softDeleteVertices(String uniq, String type) {
    return updateJanusGraphVertexDeletedStatus(uniq, type, true);
  }

  @Override
  public boolean recoverDeletedVertices(String uniq, String type) {
    return updateJanusGraphVertexDeletedStatus(uniq, type, false);
  }

  @Override
  public boolean hardDeleteVertices(String uniq, String type) {
    if (StringUtils.isNotBlank(uniq)) {
      try {
        janusGraph.traversal().V().has(VERTEX_PROPERTY_UNIQ, uniq).drop().iterate();
        janusGraph.tx().commit();
        return true;
      } catch (Throwable t) {
        logger.error(t.getMessage(), t);
        janusGraph.tx().rollback();
        return false;
      }
    }
    if (StringUtils.isNotBlank(type)) {
      try {
        janusGraph.traversal().V().has(VERTEX_PROPERTY_TYPE, type).drop().iterate();
        janusGraph.tx().commit();
        return true;
      } catch (Throwable t) {
        logger.error(t.getMessage(), t);
        janusGraph.tx().rollback();
        return false;
      }
    }
    try {
      janusGraph.traversal().V().drop().iterate();
      janusGraph.tx().commit();
      return true;
    } catch (Throwable t) {
      logger.error(t.getMessage(), t);
      janusGraph.tx().rollback();
      return false;
    }
  }

  private Edge getJanusGraphEdgeById(String edgeId) {
    if (StringUtils.isBlank(edgeId)) {
      return null;
    }
    return janusGraph.traversal().E(edgeId).next();
  }

  private GraphTraversal<Edge, Edge> getJanusGraphEdgeTraversal(
      String uniq, String type, String sourceVertexUniq, String targetVertexUniq) {
    if (StringUtils.isNotBlank(uniq)) {
      return janusGraph.traversal().E().has(EDGE_PROPERTY_UNIQ, uniq);
    }
    if (StringUtils.isNotBlank(type)) {
      if (StringUtils.isNotBlank(sourceVertexUniq) && StringUtils.isNotBlank(targetVertexUniq)) {
        return janusGraph
            .traversal()
            .E()
            .has(EDGE_PROPERTY_TYPE, type)
            .has(EDGE_PROPERTY_SOURCE_VERTEX_UNIQ, sourceVertexUniq)
            .has(EDGE_PROPERTY_TARGET_VERTEX_UNIQ, targetVertexUniq);
      } else if (StringUtils.isNotBlank(sourceVertexUniq)
          && StringUtils.isBlank(targetVertexUniq)) {
        return janusGraph
            .traversal()
            .E()
            .has(EDGE_PROPERTY_TYPE, type)
            .has(EDGE_PROPERTY_SOURCE_VERTEX_UNIQ, sourceVertexUniq);
      } else if (StringUtils.isBlank(sourceVertexUniq)
          && StringUtils.isNotBlank(targetVertexUniq)) {
        return janusGraph
            .traversal()
            .E()
            .has(EDGE_PROPERTY_TYPE, type)
            .has(EDGE_PROPERTY_TARGET_VERTEX_UNIQ, targetVertexUniq);
      } else {
        return janusGraph.traversal().E().has(EDGE_PROPERTY_TYPE, type);
      }
    }
    if (StringUtils.isNotBlank(sourceVertexUniq) && StringUtils.isNotBlank(targetVertexUniq)) {
      return janusGraph
          .traversal()
          .E()
          .has(EDGE_PROPERTY_SOURCE_VERTEX_UNIQ, sourceVertexUniq)
          .has(EDGE_PROPERTY_TARGET_VERTEX_UNIQ, targetVertexUniq);
    } else if (StringUtils.isNotBlank(sourceVertexUniq) && StringUtils.isBlank(targetVertexUniq)) {
      return janusGraph.traversal().E().has(EDGE_PROPERTY_SOURCE_VERTEX_UNIQ, sourceVertexUniq);
    } else if (StringUtils.isBlank(sourceVertexUniq) && StringUtils.isNotBlank(targetVertexUniq)) {
      return janusGraph.traversal().E().has(EDGE_PROPERTY_TARGET_VERTEX_UNIQ, targetVertexUniq);
    }
    return janusGraph.traversal().E();
  }

  private List<Edge> getJanusGraphEdges(
      String uniq, String type, String sourceVertexUniq, String targetVertexUniq) {
    GraphTraversal<Edge, Edge> janusGraphEdgeTraversal =
        getJanusGraphEdgeTraversal(uniq, type, sourceVertexUniq, targetVertexUniq);
    if (null != janusGraphEdgeTraversal) {
      return janusGraphEdgeTraversal.toList();
    }
    return Collections.emptyList();
  }

  private String createJanusGraphEdge(
      String uniq, String type, Vertex sourceVertex, Vertex targetVertex, Long createTime) {
    if (StringUtils.isBlank(uniq)) {
      return null;
    }
    if (null == sourceVertex || null == targetVertex) {
      return null;
    }
    try {
      Edge edge = sourceVertex.addEdge(LABEL_EDGE, targetVertex);
      edge.property(EDGE_PROPERTY_UNIQ, uniq);
      if (StringUtils.isNotBlank(type)) {
        edge.property(EDGE_PROPERTY_TYPE, type);
      }
      edge.property(EDGE_PROPERTY_DELETED, false);
      if (sourceVertex.property(VERTEX_PROPERTY_UNIQ).isPresent()) {
        edge.property(
            EDGE_PROPERTY_SOURCE_VERTEX_UNIQ, sourceVertex.value(VERTEX_PROPERTY_UNIQ).toString());
      }
      if (targetVertex.property(VERTEX_PROPERTY_UNIQ).isPresent()) {
        edge.property(
            EDGE_PROPERTY_TARGET_VERTEX_UNIQ, targetVertex.value(VERTEX_PROPERTY_UNIQ).toString());
      }
      if (null != createTime) {
        edge.property(EDGE_PROPERTY_CREATE_TIME, createTime);
      } else {
        edge.property(EDGE_PROPERTY_CREATE_TIME, DateUtil.getCurrentTimestamp());
      }
      janusGraph.tx().commit();
      return edge.id().toString();
    } catch (Throwable t) {
      logger.error(t.getMessage(), t);
      janusGraph.tx().rollback();
      return null;
    }
  }

  @Override
  public String createEdge(
      String uniq, String type, String sourceVertexUniq, String targetVertexUniq, Long createTime) {
    if (StringUtil.checkStringParamsBlank(uniq, sourceVertexUniq, targetVertexUniq)) {
      return null;
    }
    List<Edge> edges = getJanusGraphEdges(uniq, null, null, null);
    if (null != edges && !edges.isEmpty()) {
      return null;
    }
    List<Vertex> sourceVertices = getJanusGraphVerticesByUniq(sourceVertexUniq);
    if (null == sourceVertices || sourceVertices.isEmpty()) {
      Long vertexId = createVertex(sourceVertexUniq, type, createTime);
      if (null != vertexId) {
        sourceVertices = new ArrayList<>();
        sourceVertices.add(getJanusGraphVertexById(vertexId));
      } else {
        logger.warn(
            "Create edge uniq='{}',type='{}',sourceVertexUniq='{}',targetVertexUniq='{}' failed when create source vertex.",
            uniq,
            type,
            sourceVertexUniq,
            targetVertexUniq);
        return null;
      }
    }
    List<Vertex> targetVertices = getJanusGraphVerticesByUniq(targetVertexUniq);
    if (null == targetVertices || targetVertices.isEmpty()) {
      Long vertexId = createVertex(targetVertexUniq, type, createTime);
      if (null != vertexId) {
        targetVertices = new ArrayList<>();
        targetVertices.add(getJanusGraphVertexById(vertexId));
      } else {
        logger.warn(
            "Create edge uniq='{}',type='{}',sourceVertexUniq='{}',targetVertexUniq='{}' failed when create target vertex.",
            uniq,
            type,
            sourceVertexUniq,
            targetVertexUniq);
        return null;
      }
    }
    if (sourceVertices.size() > 1) {
      logger.warn(
          "## CreateEdge method query '{}' vertices by sourceVertexUniq='{}'.",
          sourceVertices.size(),
          sourceVertexUniq);
    }
    if (targetVertices.size() > 1) {
      logger.warn(
          "## CreateEdge method query '{}' vertices by targetVertexUniq='{}'.",
          targetVertices.size(),
          targetVertexUniq);
    }
    return createJanusGraphEdge(
        uniq, type, sourceVertices.get(0), targetVertices.get(0), createTime);
  }

  @Override
  public String createEdgeNotPreCheck(
      String uniq, String type, Long sourceVertexId, Long targetVertexId, Long createTime) {
    if (StringUtils.isBlank(uniq)) {
      return null;
    }
    Vertex sourceVertex = getJanusGraphVertexById(sourceVertexId);
    if (null == sourceVertex) {
      return null;
    }
    Vertex targetVertex = getJanusGraphVertexById(targetVertexId);
    if (null == targetVertex) {
      return null;
    }
    return createJanusGraphEdge(uniq, type, sourceVertex, targetVertex, createTime);
  }

  private boolean setJanusGraphEdgesDeletedStatus(List<Edge> edges, boolean deleted) {
    if (null == edges || edges.isEmpty()) {
      return false;
    }
    for (Edge edge : edges) {
      try {
        boolean skip = true;
        if (edge.property(EDGE_PROPERTY_DELETED).isPresent()) {
          boolean current = Boolean.parseBoolean(edge.value(EDGE_PROPERTY_DELETED).toString());
          if (current != deleted) {
            edge.property(EDGE_PROPERTY_DELETED, deleted);
            skip = false;
          }
        } else {
          edge.property(EDGE_PROPERTY_DELETED, deleted);
          skip = false;
        }
        if (!skip) {
          janusGraph.tx().commit();
        }
      } catch (Throwable t) {
        logger.error(t.getMessage(), t);
        janusGraph.tx().rollback();
      }
    }
    return true;
  }

  private boolean deleteJanusGraphEdges(
      String uniq, String type, String sourceVertexUniq, String targetVertexUniq, boolean deleted) {
    List<Edge> edges = getJanusGraphEdges(uniq, type, sourceVertexUniq, targetVertexUniq);
    if (null == edges || edges.isEmpty()) {
      return false;
    }
    if (edges.size() > 1) {
      logger.warn(
          "## DeleteJanusGraphEdges method query '{}' edges by uniq={},type='{}',sourceVertexUniq={},targetVertexUniq={}.",
          edges.size(),
          uniq,
          type,
          sourceVertexUniq,
          targetVertexUniq);
    }
    return setJanusGraphEdgesDeletedStatus(edges, deleted);
  }

  @Override
  public boolean softDeleteEdges(
      String uniq, String type, String sourceVertexUniq, String targetVertexUniq) {
    return deleteJanusGraphEdges(uniq, type, sourceVertexUniq, targetVertexUniq, true);
  }

  @Override
  public boolean recoverDeletedEdges(
      String uniq, String type, String sourceVertexUniq, String targetVertexUniq) {
    return deleteJanusGraphEdges(uniq, type, sourceVertexUniq, targetVertexUniq, false);
  }

  private boolean hardDeleteJanusGraphEdges(
      String uniq, String type, String sourceVertexUniq, String targetVertexUniq) {
    GraphTraversal<Edge, Edge> janusGraphEdgeTraversal =
        getJanusGraphEdgeTraversal(uniq, type, sourceVertexUniq, targetVertexUniq);
    if (null == janusGraphEdgeTraversal) {
      return false;
    }
    try {
      janusGraphEdgeTraversal.drop().iterate();
      janusGraph.tx().commit();
    } catch (Throwable t) {
      logger.error(t.getMessage(), t);
      janusGraph.tx().rollback();
      return false;
    }
    return true;
  }

  @Override
  public boolean hardDeleteEdges(
      String uniq, String type, String sourceVertexUniq, String targetVertexUniq) {
    return hardDeleteJanusGraphEdges(uniq, type, sourceVertexUniq, targetVertexUniq);
  }

  private GraphVertex transferToGraphVertex(Vertex v) {
    if (null == v) {
      return null;
    }
    //    org.studiox.graph.janusgraph.common.JanusGraphVertex graphVertex =
    //        new org.studiox.graph.janusgraph.common.JanusGraphVertex();
    GraphVertex graphVertex = new GraphVertex();
    try {
      //      graphVertex.setId(Long.parseLong(v.id().toString()));
      if (v.property(VERTEX_PROPERTY_UNIQ).isPresent()) {
        graphVertex.setUniq(v.value(VERTEX_PROPERTY_UNIQ).toString());
      }
      if (v.property(VERTEX_PROPERTY_TYPE).isPresent()) {
        graphVertex.setType(v.value(VERTEX_PROPERTY_TYPE).toString());
      }
      if (v.property(VERTEX_PROPERTY_DELETED).isPresent()) {
        graphVertex.setDeleted(Boolean.valueOf(v.value(VERTEX_PROPERTY_DELETED).toString()));
      }
      if (v.property(VERTEX_PROPERTY_CREATE_TIME).isPresent()) {
        graphVertex.setCreateTime(Long.parseLong(v.value(VERTEX_PROPERTY_CREATE_TIME).toString()));
      }
    } catch (Throwable t) {
      logger.error(t.getMessage(), t);
    }
    return graphVertex;
  }

  @Override
  public Long getJanusGraphVertexId(String uniq) {
    List<Vertex> vertices = getJanusGraphVerticesByUniq(uniq);
    if (null == vertices || vertices.isEmpty()) {
      return null;
    }
    if (vertices.size() > 1) {
      logger.warn("## GetVertex method query '{}' vertices by uniq='{}'.", vertices.size(), uniq);
    }
    return Long.parseLong(vertices.get(0).id().toString());
  }

  @Override
  public GraphVertex getVertex(String uniq) {
    List<Vertex> vertices = getJanusGraphVerticesByUniq(uniq);
    if (null == vertices || vertices.isEmpty()) {
      return null;
    }
    if (vertices.size() > 1) {
      logger.warn("## GetVertex method query '{}' vertices by uniq='{}'.", vertices.size(), uniq);
    }
    return transferToGraphVertex(vertices.get(0));
  }

  @Override
  public List<GraphVertex> getVerticesByType(String type) {
    List<Vertex> vertices = getJanusGraphVerticesByType(type);
    if (null == vertices || vertices.isEmpty()) {
      return Collections.emptyList();
    }
    List<GraphVertex> result = new ArrayList<>();
    for (Vertex v : vertices) {
      GraphVertex vertex = transferToGraphVertex(v);
      if (null == vertex) {
        continue;
      }
      result.add(vertex);
    }
    return result;
  }

  private GraphEdge transferToGraphEdge(Edge e) {
    if (null == e) {
      return null;
    }
    //    org.studiox.graph.janusgraph.common.JanusGraphEdge graphEdge =
    //        new org.studiox.graph.janusgraph.common.JanusGraphEdge();
    GraphEdge graphEdge = new GraphEdge();
    try {
      //      graphEdge.setId(((RelationIdentifier) e.id()).toString());
      //      graphEdge.setSourceVertexId(Long.parseLong(e.outVertex().id().toString()));
      //      graphEdge.setTargetVertexId(Long.parseLong(e.inVertex().id().toString()));
      if (e.property(EDGE_PROPERTY_UNIQ).isPresent()) {
        graphEdge.setUniq(e.value(EDGE_PROPERTY_UNIQ).toString());
      }
      if (e.property(EDGE_PROPERTY_TYPE).isPresent()) {
        graphEdge.setType(e.value(EDGE_PROPERTY_TYPE).toString());
      }
      if (e.property(EDGE_PROPERTY_DELETED).isPresent()) {
        graphEdge.setDeleted(Boolean.valueOf(e.value(EDGE_PROPERTY_DELETED).toString()));
      }
      if (e.property(EDGE_PROPERTY_SOURCE_VERTEX_UNIQ).isPresent()) {
        graphEdge.setSourceVertexUniq(e.value(EDGE_PROPERTY_SOURCE_VERTEX_UNIQ).toString());
      }
      if (e.property(EDGE_PROPERTY_TARGET_VERTEX_UNIQ).isPresent()) {
        graphEdge.setTargetVertexUniq(e.value(EDGE_PROPERTY_TARGET_VERTEX_UNIQ).toString());
      }
      if (e.property(EDGE_PROPERTY_CREATE_TIME).isPresent()) {
        graphEdge.setCreateTime(Long.parseLong(e.value(EDGE_PROPERTY_CREATE_TIME).toString()));
      }
    } catch (Throwable t) {
      logger.error(t.getMessage(), t);
    }
    return graphEdge;
  }

  private List<GraphEdge> transferToGraphEdges(List<Edge> edges) {
    if (null == edges || edges.isEmpty()) {
      return Collections.emptyList();
    }
    List<GraphEdge> result = new ArrayList<>();
    for (Edge e : edges) {
      GraphEdge edge = transferToGraphEdge(e);
      if (null == e) {
        continue;
      }
      result.add(edge);
    }
    return result;
  }

  @Override
  public GraphEdge getEdge(String uniq) {
    List<Edge> edges = getJanusGraphEdges(uniq, null, null, null);
    if (null == edges || edges.isEmpty()) {
      return null;
    }
    return transferToGraphEdge(edges.get(0));
  }

  @Override
  public List<GraphEdge> getEdges(String type, String sourceVertexUniq, String targetVertexUniq) {
    List<Edge> edges = getJanusGraphEdges(null, type, sourceVertexUniq, targetVertexUniq);
    return transferToGraphEdges(edges);
  }

  @Override
  public Map<GraphEdge, GraphVertex> getSourceVertexEdgeMap(String vertexUniq) {
    if (StringUtils.isBlank(vertexUniq)) {
      return Collections.emptyMap();
    }
    List<Edge> edges =
        janusGraph.traversal().V().has(VERTEX_PROPERTY_UNIQ, vertexUniq).inE(LABEL_EDGE).toList();
    if (null == edges || edges.isEmpty()) {
      return Collections.emptyMap();
    }
    Map<GraphEdge, GraphVertex> result = new HashMap<>(edges.size());
    for (Edge e : edges) {
      GraphEdge graphEdge = transferToGraphEdge(e);
      GraphVertex graphVertex = transferToGraphVertex(e.inVertex());
      result.put(graphEdge, graphVertex);
    }
    return result;
  }

  @Override
  public Map<GraphEdge, GraphVertex> getTargetVertexEdgeMap(String vertexUniq) {
    if (StringUtils.isBlank(vertexUniq)) {
      return Collections.emptyMap();
    }
    List<Edge> edges =
        janusGraph.traversal().V().has(VERTEX_PROPERTY_UNIQ, vertexUniq).outE(LABEL_EDGE).toList();
    if (null == edges || edges.size() == 0) {
      return Collections.emptyMap();
    }
    Map<GraphEdge, GraphVertex> result = new HashMap<>(edges.size());
    for (Edge e : edges) {
      GraphEdge graphEdge = transferToGraphEdge(e);
      GraphVertex graphVertex = transferToGraphVertex(e.outVertex());
      result.put(graphEdge, graphVertex);
    }
    return result;
  }

  private GraphLineage transferToGraphLineage(String vertexUniq, Graph graph) {
    if (StringUtils.isBlank(vertexUniq) || null == graph) {
      return null;
    }
    GraphVertex currentVertex = getVertex(vertexUniq);
    GraphLineage graphLineage = new GraphLineage();
    graphLineage.setCurrent(currentVertex);
    graphLineage.setVertices(new ArrayList<>());
    graphLineage.setEdges(new ArrayList<>());
    Iterator<Vertex> vertexIterator = graph.vertices();
    while (vertexIterator.hasNext()) {
      Vertex v = vertexIterator.next();
      graphLineage.getVertices().add(transferToGraphVertex(v));
    }
    Iterator<Edge> edgeIterator = graph.edges();
    while (edgeIterator.hasNext()) {
      Edge e = edgeIterator.next();
      graphLineage.getEdges().add(transferToGraphEdge(e));
    }
    return graphLineage;
  }

  @Override
  public GraphLineage getImpact(String vertexUniq, Integer depth, boolean skipDeleted) {
    if (StringUtils.isBlank(vertexUniq)) {
      return null;
    }
    GraphTraversal graphTraversal =
        janusGraph.traversal().V().has(VERTEX_PROPERTY_UNIQ, vertexUniq);
    if (skipDeleted) {
      graphTraversal = graphTraversal.has(VERTEX_PROPERTY_DELETED, false);
    }
    Graph graph;
    if (depth <= 0) {
      if (!skipDeleted) {
        graph =
            (Graph)
                graphTraversal
                    .until(__.outE(LABEL_EDGE).count().is(0))
                    .repeat(__.outE(LABEL_EDGE).subgraph("subGraph").inV().simplePath())
                    .cap("subGraph")
                    .next();
      } else {
        graph =
            (Graph)
                graphTraversal
                    .until(
                        __.outE(LABEL_EDGE)
                            .or(
                                __.has(EDGE_PROPERTY_DELETED, true),
                                __.inV().has(VERTEX_PROPERTY_DELETED, true),
                                __.count().is(0)))
                    .repeat(__.outE(LABEL_EDGE).subgraph("subGraph").inV().simplePath())
                    .cap("subGraph")
                    .next();
      }
    } else {
      if (!skipDeleted) {
        graph =
            (Graph)
                graphTraversal
                    .repeat(__.outE(LABEL_EDGE).subgraph("subGraph").inV().simplePath())
                    .times(depth)
                    .cap("subGraph")
                    .next();
      } else {
        graph =
            (Graph)
                graphTraversal
                    .repeat(
                        __.outE(LABEL_EDGE)
                            .and(
                                __.has(EDGE_PROPERTY_DELETED, false),
                                __.inV().has(VERTEX_PROPERTY_DELETED, false))
                            .subgraph("subGraph")
                            .inV()
                            .simplePath())
                    .times(depth)
                    .cap("subGraph")
                    .next();
      }
    }
    return transferToGraphLineage(vertexUniq, graph);
  }

  @Override
  public GraphLineage getLineage(String vertexUniq, Integer depth, boolean skipDeleted) {
    if (StringUtils.isBlank(vertexUniq)) {
      return null;
    }
    GraphTraversal graphTraversal =
        janusGraph.traversal().V().has(VERTEX_PROPERTY_UNIQ, vertexUniq);
    if (skipDeleted) {
      graphTraversal = graphTraversal.has(VERTEX_PROPERTY_DELETED, false);
    }
    Graph graph;
    if (depth <= 0) {
      if (!skipDeleted) {
        graph =
            (Graph)
                graphTraversal
                    .until(__.inE(LABEL_EDGE).count().is(0))
                    .repeat(__.inE(LABEL_EDGE).subgraph("subGraph").outV().simplePath())
                    .cap("subGraph")
                    .next();
      } else {
        graph =
            (Graph)
                graphTraversal
                    .until(
                        __.inE(LABEL_EDGE)
                            .or(
                                __.has(EDGE_PROPERTY_DELETED, true),
                                __.outV().has(VERTEX_PROPERTY_DELETED, true),
                                __.count().is(0)))
                    .repeat(__.inE(LABEL_EDGE).subgraph("subGraph").outV().simplePath())
                    .cap("subGraph")
                    .next();
      }
    } else {
      if (!skipDeleted) {
        graph =
            (Graph)
                graphTraversal
                    .repeat(__.inE(LABEL_EDGE).subgraph("subGraph").outV().simplePath())
                    .times(depth)
                    .cap("subGraph")
                    .next();
      } else {
        graph =
            (Graph)
                graphTraversal
                    .repeat(
                        __.inE(LABEL_EDGE)
                            .and(
                                __.has(EDGE_PROPERTY_DELETED, false),
                                __.outV().has(VERTEX_PROPERTY_DELETED, false))
                            .subgraph("subGraph")
                            .outV()
                            .simplePath())
                    .times(depth)
                    .cap("subGraph")
                    .next();
      }
    }
    return transferToGraphLineage(vertexUniq, graph);
  }

  @Override
  public GraphLineage getFullLineage(String vertexUniq, Integer depth, boolean skipDeleted) {
    if (StringUtils.isBlank(vertexUniq)) {
      return null;
    }
    GraphTraversal graphTraversal =
        janusGraph.traversal().V().has(VERTEX_PROPERTY_UNIQ, vertexUniq);
    if (skipDeleted) {
      graphTraversal = graphTraversal.has(VERTEX_PROPERTY_DELETED, false);
    }
    Graph graph;
    if (depth <= 0) {
      if (!skipDeleted) {
        graph =
            (Graph)
                graphTraversal
                    .union(
                        __.until(__.outE(LABEL_EDGE).count().is(0))
                            .repeat(
                                (Traversal)
                                    __.outE(LABEL_EDGE).subgraph("subGraph").inV().simplePath()),
                        __.until(__.inE(LABEL_EDGE).count().is(0))
                            .repeat(
                                (Traversal)
                                    __.inE(LABEL_EDGE).subgraph("subGraph").outV().simplePath()))
                    .cap("subGraph")
                    .next();
      } else {
        graph =
            (Graph)
                graphTraversal
                    .union(
                        __.until(
                                __.outE(LABEL_EDGE)
                                    .or(
                                        __.has(EDGE_PROPERTY_DELETED, true),
                                        __.inV().has(VERTEX_PROPERTY_DELETED, true),
                                        __.count().is(0)))
                            .repeat(
                                (Traversal)
                                    __.outE(LABEL_EDGE).subgraph("subGraph").inV().simplePath()),
                        __.until(
                                __.inE(LABEL_EDGE)
                                    .or(
                                        __.has(EDGE_PROPERTY_DELETED, true),
                                        __.outV().has(VERTEX_PROPERTY_DELETED, true),
                                        __.count().is(0)))
                            .repeat(
                                (Traversal)
                                    __.inE(LABEL_EDGE).subgraph("subGraph").outV().simplePath()))
                    .cap("subGraph")
                    .next();
      }
    } else {
      if (!skipDeleted) {
        graph =
            (Graph)
                graphTraversal
                    .union(
                        __.repeat(__.outE(LABEL_EDGE).subgraph("subGraph").inV().simplePath())
                            .times(depth),
                        __.repeat(__.inE(LABEL_EDGE).subgraph("subGraph").outV().simplePath())
                            .times(depth))
                    .cap("subGraph")
                    .next();

      } else {
        graph =
            (Graph)
                graphTraversal
                    .union(
                        __.repeat(
                                __.outE(LABEL_EDGE)
                                    .and(
                                        __.has(EDGE_PROPERTY_DELETED, false),
                                        __.inV().has(VERTEX_PROPERTY_DELETED, false))
                                    .subgraph("subGraph")
                                    .inV()
                                    .simplePath())
                            .times(depth),
                        __.repeat(
                                __.inE(LABEL_EDGE)
                                    .and(
                                        __.has(EDGE_PROPERTY_DELETED, false),
                                        __.outV().has(VERTEX_PROPERTY_DELETED, false))
                                    .subgraph("subGraph")
                                    .outV()
                                    .simplePath())
                            .times(depth))
                    .cap("subGraph")
                    .next();
      }
    }
    return transferToGraphLineage(vertexUniq, graph);
  }
}
